package com.study.kafka.service;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.kafka.annotation.PartitionOffset;
import org.springframework.kafka.annotation.TopicPartition;
import org.springframework.kafka.config.KafkaListenerEndpointRegistry;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.kafka.support.Acknowledgment;
import org.springframework.stereotype.Service;

/**
 * @author flwu
 * @date 2020/3/19 7:21 PM
 * @MOTTO 保持空杯和敬畏
 */
@Service
public class ConsumerService {

    @Autowired
    private KafkaTemplate kafkaTemplate;

    /**
     * 监听器生命周期管理
     */
    @Autowired
    private KafkaListenerEndpointRegistry registry;

    /**
     * 获取数据--最复杂的用法
     * @param concurrency 设置并发量， 小于或等于Topic的分区数
     * @param partitionOffsets 这是偏移量
     * @param message
     */
    @KafkaListener(id = "test", topicPartitions = {
            @TopicPartition(topic = "lubai", partitions = {"0", "1"}),
            @TopicPartition(topic = "ctrip", partitions = "0",
                    partitionOffsets = @PartitionOffset(partition = "1", initialOffset = "1000"))
    }, concurrency = "6", errorHandler = "myErrorHandler")
    public void receive(String message){
        System.out.println(message);
    }

    /**
     * 开启ACK开启时，需要在方法里面手动ACK
     * @param message
     * @param ack
     */
    @KafkaListener(id = "test1", topics = {"lubai", "ctrip"})
    public void receive(String message, Acknowledgment ack) {
        System.out.println(message);
        //手动ack
        ack.acknowledge();
    }

    /**
     * 监听器的生命周期管理
     * @KafkaListener 只是注册监听到KafkaListenerEndpointRegistry， 并不会自己执行，需要KafkaListenerEndpointRegistry控制执行
     */
    public void monitor() {
        //开始
        registry.getListenerContainer("test").start();
        //暂停
        registry.getListenerContainer("test").pause();
        //恢复
        registry.getListenerContainer("test").resume();
    }

}
